package com.atguigu.edu.realtime.app.dwd.db;

import com.atguigu.edu.realtime.utils.MyKafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

//注册
public class DwdUserEnroll {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(5);

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        tableEnv.executeSql(MyKafkaUtil.getTopicDbDDL("DwdUserEnroll"));
        //  tableEnv.executeSql("select * from topic_db").print();
        Table table = tableEnv.sqlQuery("select `data`['id'] id,`data`['create_time'] create_time from topic_db where `table`='user_info'");

        tableEnv.createTemporaryView("user_info", table);

        tableEnv.executeSql("CREATE TABLE dwd_user_enroll (\n" +
                "  id string,\n" +
                "  create_time string,\n" +
                "  PRIMARY KEY (id) NOT ENFORCED\n" +
                ")" + MyKafkaUtil.getUpsertKafkaDDL("dwd_user_enroll"));
        //5.2 写入
        tableEnv.executeSql("insert into dwd_user_enroll select * from user_info");


    }
}
